package com.xzx.flink.streamapi.sink;

import com.xzx.flink.bean.ClickEvent;
import com.xzx.flink.streamapi.source.ClickSource;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;

import java.util.concurrent.TimeUnit;

/**
 * @version 1.0
 * @auther xinzhixuan
 * @date 2022/4/18 12:05
 */
public class Sink_01_File_Row_Encode {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<ClickEvent> source = env.addSource(new ClickSource());

        StreamingFileSink<String> fileSink = StreamingFileSink.forRowFormat(new Path("file:///Users/xinzhixuan/work/git/flink-study-java/out"), new SimpleStringEncoder<String>("UTF-8"))
                .withRollingPolicy(DefaultRollingPolicy.builder()//在以下 3 种情况下，我们就会滚动分区文件
                        .withRolloverInterval(TimeUnit.MINUTES.toMinutes(15))//至少包含 15 分钟的数据
                        .withInactivityInterval(TimeUnit.MINUTES.toMinutes(5))//最近 5 分钟没有收到新的数据
                        .withMaxPartSize(1024 * 1024 * 1024)//文件大小已达到 1 GB
                        .build())
                .build();
        source.map(ClickEvent::toString).addSink(fileSink);
        env.execute(Sink_01_File_Row_Encode.class.getSimpleName());
    }
}
